package com.udf.flink.scala.examples


import java.time.Duration
import com.udf.flink.udsf.TsToLong
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala.function.{ProcessWindowFunction, WindowFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer
import scala.collection.parallel.immutable

object keyby_datastream {


  case class UserGrade(user_id: Int, weekday: String, grade: Int)
  case class User(user_id: Int, grade: Int, a: String,  rowtime: Long)
  case class UserGradeList(user_id: Int, gradeList: List[Int])
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // set up the Scala Table API
    val tableEnv = StreamTableEnvironment.create(env)
    val sql= """CREATE TABLE ordr (
              id INT,
              cnt INT,
              user_id INT,
              a STRING,
              rowtime AS localtimestamp,
              WATERMARK FOR rowtime AS rowtime
              ) WITH (
                'connector' = 'datagen',
              'rows-per-second'='1',
              'fields.id.kind'='sequence',
              'fields.id.start'='1',
              'fields.id.end'='200',
              'fields.cnt.min'='1',
              'fields.cnt.max'='2',
              'fields.user_id.min'='1',
              'fields.user_id.max'='20',
              'fields.a.length'='1'
              )"""
    tableEnv.executeSql(sql)
    tableEnv.createTemporarySystemFunction("TsToLong", classOf[TsToLong])
    val tb=tableEnv.sqlQuery("""select user_id,cnt,a,TsToLong(rowtime) rowtime from ordr""")

//    val socketStream: DataStream[String] = env.socketTextStream("localhost", 8888)

//    val userGradeStream: DataStream[UserGrade] = socketStream.map((data: String) => {
//      val user_array: Array[String] = data.split(",")
//      UserGrade(user_array(0), user_array(1), user_array(2).toInt)
//    })

    val keyWondowStream: WindowedStream[User, Int, TimeWindow] = tableEnv.toDataStream(tb,classOf[User])
       .assignTimestampsAndWatermarks( WatermarkStrategy
          .forBoundedOutOfOrderness(Duration.ofSeconds(5))
          .withTimestampAssigner(new SerializableTimestampAssigner[User]
            {
                override def extractTimestamp(element: User,recordTimestamp: Long): Long = element.rowtime
            })
       )
      .keyBy((_: User).user_id)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))


    //    Base interface for functions that are evaluated over keyed (grouped) windows.
    //    trait WindowFunction[IN, OUT, KEY, W <: Window] extends Function with Serializable
//        val resultStream: DataStream[(Int, List[Int])] = keyWondowStream.apply(new WindowFunction[User, (Int, List[Int]), Int, TimeWindow] {
//          override def apply(key: Int, window: TimeWindow, input: Iterable[User], out: Collector[(Int, List[Int])]): Unit = {
//            var gradeList: ListBuffer[Int] = ListBuffer[Int]()
//            for (data <- input) {
//              gradeList += data.grade
//            }
//            out.collect((key,gradeList.toList.sorted))
//          }
//        })


    //        val resultStream: DataStream[String] = keyWondowStream.apply(new WindowFunction[UserGrade, String, String, TimeWindow] {
    //          override def apply(key: String, window: TimeWindow, input: Iterable[UserGrade], out: Collector[String]): Unit = {
    //            var gradeList: ListBuffer[Int] = ListBuffer[Int]()
    //            for (data <- input){
    //              gradeList+=data.grade
    //            }
    //            val result: String = key + "===" + gradeList.toList.sorted.toString
    //            out.collect(result)
    //          }
    //        })


    val resultStream: DataStream[UserGradeList] = keyWondowStream.process(new ProcessWindowFunction[User, UserGradeList, Int, TimeWindow]{
      override def process(key: Int, context: Context, elements: Iterable[User], out: Collector[UserGradeList]): Unit = {
        var gradeList: ListBuffer[Int] = ListBuffer[Int]()
        for (data <- elements) {
          gradeList += data.grade

        }

        out.collect(UserGradeList(key,gradeList.toList.sorted))
      }
    })

    resultStream.print("===")
    env.execute("process apply function")
  }
}